package com.xjscrm.console.mq.consumer;

import com.uzai.trace.TraceGenerate;
import com.xjscrm.console.dto.circle.CircleTaskMqDTO;
import com.xjscrm.console.entity.CircleTask;
import com.xjscrm.console.queue.DelayProducer;
import com.xjscrm.console.queue.DelayTask;
import com.xjscrm.console.service.customer.circle.CircleTaskService;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.ObjectUtils;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.TimeUnit;

@RocketMQMessageListener(topic = "xj_pull_circle_task_topic", consumerGroup = "xj_pull_circle_task_group", messageModel = MessageModel.CLUSTERING)
@Component
@Slf4j
public class PullCircleTaskMessageConsumer implements RocketMQListener<CircleTaskMqDTO> {

    @Autowired
    private RedisTemplate<Object, Object> redisTemplate;
    @Autowired
    private CircleTaskService circleTaskService;
    @Autowired
    private RocketMQTemplate rocketMQTemplate;
    @Autowired
    private Executor chatExecutor;

    public static Runnable sendCircleTask(RocketMQTemplate rocketMQTemplate, CircleTaskMqDTO circleTaskMqDTO) {

        return () -> {
            rocketMQTemplate.asyncSend("xj_circle_send_topic", MessageBuilder.withPayload(circleTaskMqDTO).build(), new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                    if (log.isDebugEnabled()) {
                        log.debug("分发-发送朋友圈任务的消息成功: {}", circleTaskMqDTO);
                    }
                }

                @Override
                public void onException(Throwable e) {
                    log.error("分发-发送朋友圈任务的消息失败", e);
                }
            });
        };
    }

    @Override
    @TraceGenerate
    public void onMessage(CircleTaskMqDTO taskMq) {

        log.info("收到拉取朋友圈任务消息: {}", taskMq);

        Long merId = taskMq.getMerId();

        String mer_circle_task_flag = "xj_pull_circle_task_flag_%s";
        String format = String.format(mer_circle_task_flag, merId);
        //
        if (Boolean.FALSE.equals(redisTemplate.opsForValue().setIfAbsent(format, merId, 60, TimeUnit.SECONDS))) {
            log.error("消息已被其他节点处理: {}", taskMq);
            return;
        }

        try {
            List<CircleTask> circleTaskList = circleTaskService.listValidByMerId(merId);

            if (ObjectUtils.isEmpty(circleTaskList)) {
                log.info("未找到朋友圈任务, 程序终止: {}", taskMq);
                return;
            }

            Integer current = (int) (System.currentTimeMillis() / 1000L);

            for (CircleTask circleTask : circleTaskList) {

                CircleTaskMqDTO circleTaskMqDTO = new CircleTaskMqDTO();
                circleTaskMqDTO.setMerId(circleTask.getMerId());
                circleTaskMqDTO.setTaskId(circleTask.getId());

                Integer executeTime = circleTask.getExecuteTime();
                if (executeTime <= current) {
                    log.info("执行时间小于当前时间, 立即执行. 任务id: {}", circleTask.getId());
                    chatExecutor.execute(sendCircleTask(rocketMQTemplate, circleTaskMqDTO));
                    continue;
                }

                DelayTask delayTask = new DelayTask(merId, circleTask.getId(), circleTask.getName(), executeTime.longValue() - current.longValue(), sendCircleTask(rocketMQTemplate, circleTaskMqDTO));
                DelayProducer.instance.addSendDelayTask(delayTask);
            }
        } finally {
            redisTemplate.expire(format, 0, TimeUnit.MILLISECONDS);
        }
    }
}